package com.huqing.icu.service.app;

import com.huqing.icu.constants.KafkaTopicConstants;
import com.huqing.icu.mapper.CommunityPostMapper;
import com.huqing.icu.utils.JsonUtils;
import jakarta.annotation.Resource;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

import java.util.Optional;

/**
 * @Description 社区帖子消费者
 * @Author huqing
 * @Date 2025/5/31 14:12
 **/
@Service
public class CommunityPostConsumer {

    @Resource
    private CommunityPostMapper communityPostMapper;

    private static final Logger logger = LoggerFactory.getLogger(CommunityPostConsumer.class);

    @KafkaListener(topics = KafkaTopicConstants.DELETE_COMMUNITY_POST, groupId = "TEST_GROUP", concurrency = "2", containerFactory = "kafkaManualAckListenerContainerFactory")
    public void deletePost(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            Object msg = message.get();
            try {
                // 这里写你对接收到的消息的处理逻辑
                // 走消息发送逻辑
                Long postId = JsonUtils.getClazz(msg.toString(), Long.class);
                communityPostMapper.deleteById(postId);
                // 手动ACK
                ack.acknowledge();
                logger.info("Kafka消费: 删除社区帖子成功! Topic:" + topic + ",Message:" + msg);
            } catch (Exception e) {
                logger.error("Kafka消费: 删除社区帖失败！Topic:" + topic + ",Message:" + msg, e);
            }
        }
    }
}
